package com.hujf.flink.DataStreamAPI;

import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.configuration.Configuration;

/**
 * @Author hujf
 * @Date 2022/5/18 16:37
 * @Description TODO
 */
public class ConfigDemo {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment environment = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<Integer> integerDataSource = environment.fromElements(1, 2, 3, 54, 5, 2, 3, 4);

        Configuration configuration = new Configuration();
        configuration.setInteger("limit",10);

        integerDataSource.filter(new RichFilterFunction<Integer>() {
            private Integer limit;
            @Override
            public boolean filter(Integer value) throws Exception {
                return value>limit;
            }

            @Override
            public void open(Configuration parameters) throws Exception {
                limit = parameters.getInteger("limit",0);
            }
        }).withParameters(configuration).print();
    }
}
